import java.io.IOException
import java.sql.Timestamp
import java.util.Properties

import org.apache.spark.sql.functions._
import org.apache.spark.{SparkConf, SparkContext}
import common.beans.{RecordBean, Stock}
import config.ConfigurationFactory
import utils.JsonUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.Logger
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext, SaveMode, SparkSession}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferBrokers
import org.apache.spark.streaming.{Seconds, StreamingContext}
import spray.json._

import scala.reflect.ClassTag

object KafkaStream {
  private[this] lazy val logger = Logger.getLogger(getClass)

  private[this] val config = ConfigurationFactory.load()

  /**
    * Json decode UDF function
    *
    * @param text the encoded JSON string
    * @return Returns record bean
    */
  def jsonDecode(text: String): RecordBean = {
    try {
      JsonUtils.deserialize(text, classOf[RecordBean])
    } catch {
      case e: IOException =>
        logger.error(e.getMessage, e)
        null
    }
  }

  def main(args: Array[String]): Unit = {
//    val spark = SparkSession.builder
//      .appName("spark-kafka-stock")
//      // .master("spark://sparkmaster:7077")
//      .master("yarn")
//      .getOrCreate
//
//    println("..........spark:"+spark)
//    println("..........config"+config.window)
//    val streaming = new StreamingContext(spark.sparkContext, Seconds(5))
  val conf = new SparkConf()
  .setMaster("yarn")
  .setAppName("kafkastream")
  .set("spark.executor.memory", "1g")
  .set("spark.rdd.compress","true")
  .set("spark.storage.memoryFraction", "1")
  .set("spark.streaming.unpersist", "true")
  .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
  .set("spark.kryo.registrator","MyRegistrator")

    val sc = new SparkContext(conf)
    val streaming = new StreamingContext(sc, Seconds(5))

    // val servers = config.producer;
    val servers="kafkabroker1:9092,kafkabroker2:9092"
    println("..............servers"+servers)
    
    val params = Map[String, Object](
      "bootstrap.servers" -> servers,
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
//      "auto.offset.reset" -> "frombeginning",
      "group.id" -> "dashboard",
      // "partition.assignment.strategy"->"range",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    // topic names which will be read
    val topics = Array("stockrealtime")

    // create kafka direct stream object
    val stream = KafkaUtils.createDirectStream[String, String](
      streaming, PreferBrokers, Subscribe[String, String](topics, params))

    // our table has 3 fields called market (varchar), rate (float) and dt (datetime etc.)
    val schema = StructType(
        StructField("code", StringType) ::
        StructField("topen", FloatType) ::
        StructField("timestr", FloatType) ::
        StructField("yclose", FloatType) ::
        StructField("current", FloatType) ::
        StructField("tmax", FloatType) ::
        StructField("tmin", FloatType) :: Nil
    )

    // just alias for simplicity
    type Record = ConsumerRecord[String, String]
    val spark: SparkSession = SparkSession.builder
      .config(conf)
      .getOrCreate;
    import spark.implicits._

    stream.foreachRDD((rdd: RDD[Record]) => {

      //      if(rdd.count()>0) {
//
//        // convert string to PoJo and generate rows as tuple group
//        rdd.take(3).foreach(x => {
//          val jsonAst = x.value().parseJson
//          println("....."+jsonAst.asJsObject.getFields("data")(0))
////          println("....."+jsonAst.asJsObject.getFields("data").getClass())
////          println("....."+ClassTag(jsonAst.asJsObject.getFields("data").getClass()))
//        })
//      }

      if(rdd.count()>0) {
      println(".............."+rdd.count())
      rdd.collect().foreach(x=> {
        val jsonAst = x.value().parseJson

        val str=jsonAst.asJsObject.getFields("data")(0).toString()
        for(elem:Char <- str)
          println(elem)

        val values:Array[String] = jsonAst.asJsObject.getFields("data")(0).toString().split('$');
        for(elem:String <- values)
          println("..........."+elem)

        Stock(values(0), values(1), values(2), values(3), values(4), values(5), values(6), values(7))
      })
//        .take(3).foreach(println)
//        .groupBy("code").agg(collect_list("topen") as "list").show()
      }


      /**
        * aggregate data by market type
        *
        * tuple has 3 items,
        * the first one is counter value and this value is 1,
        * second one is the rate and received from Kafka,
        * third one is event time. for instance `2017-05-12 16:00:00`
        *
        * in the map,
        * method <code>f._1</code> is market name,
        * we divide total rate to total item count <code>f._2._2 / f._2._1</code>
        * as you can see <code>f._2._3</code> is average event time
        **/
    //   val flatten = pairs
    //     .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2, (y._3 + x._3) / 2))
    //     .map(f => Row.fromSeq(Seq(f._1, f._2._2 / f._2._1, new Timestamp(f._2._3))))

      // create sql context from active spark context
    //   val sql = new SQLContext(flatten.sparkContext)

      // write aggregated results to database
      // only one partition required for better visualisation,
      // better to look at https://goo.gl/iBdNDl
    //   sql.createDataFrame(flatten, schema)
    //     .repartition(1)
    //     .write
    //     .mode(SaveMode.Append)
    //     .jdbc(url, table, props)
    })

    // create streaming context and submit streaming jobs
    streaming.start()

    // wait to killing signals etc.
    streaming.awaitTermination()
  }
}